package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.code.common.Constant;
import org.code.domain.OrderEntity;
import org.code.util.ByteStringConverter;
import org.code.util.DateUtil;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

/**
 * @author xiaojian
 * <p>
 * 顺序消息
 */
public class Order {
    static final String TOPIC = "order-topic";

    @Test
    public void orderProducer() throws Exception {
        // 创建消费者
        DefaultMQProducer producer = new DefaultMQProducer("order-producer-group");
        // 设置NameServer地址
        producer.setNamesrvAddr(Constant.NAME_SERVER);
        // 启动消费者
        producer.start();
        // 订单集合
        List<OrderEntity> orderList = Arrays.asList(
                new OrderEntity(1, 111, 59D, DateUtil.getNow(), "下订单"),
                new OrderEntity(2, 111, 59D, DateUtil.getNow(), "物流"),
                new OrderEntity(3, 111, 59D, DateUtil.getNow(), "签收"),
                new OrderEntity(4, 112, 89D, DateUtil.getNow(), "下订单"),
                new OrderEntity(5, 112, 89D, DateUtil.getNow(), "物流"),
                new OrderEntity(6, 112, 89D, DateUtil.getNow(), "拒收"));

        // 循环发送消息
        for (OrderEntity order : orderList) {
            Message message = new Message(TOPIC, ByteStringConverter.stringToByteArray(order.toString()));
            producer.send(message, new MessageQueueSelector() {
                // 发送的时候 相同的订单号选择同一个队列
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    // 当前主题有多少个队列
                    int size = mqs.size();
                    // 这个arg就是后面传入的 order.getOrderNumber()
                    int orderNumber = Integer.parseInt(arg.toString());
                    // 用这个值去%队列的个数得到一个队列
                    int index = orderNumber % size;
                    // 返回选择的这个队列即可 ，那么相同的订单号 就会被放在相同的队列里 实现FIFO了
                    return mqs.get(index);
                }
            }, order.getOrderNumber());
        }
        // 关闭生产者
        producer.shutdown();
    }


    @Test
    public void orderConsumer() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
        // 设置NameServer地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER);
        // 订阅主题
        consumer.subscribe(TOPIC, "*");
        // 注册监听器
        consumer.registerMessageListener(new MessageListenerOrderly() {
            /**
             *  MessageListenerOrderly 是顺序消费 单线程消费
             */
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
                for (MessageExt messageExt : list) {
                    System.out.println(ThreadUtil.getCurrentThreadName() + " 消费消息：" +
                            ByteStringConverter.byteArrayToString(messageExt.getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 阻塞
        ThreadUtil.read();
    }


}
